package com.hgw.easyrocketmq.base.consumer.impl;

import com.hgw.easyrocketmq.base.consumer.BaseMsgConsumer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

/**
 * Description:
 *
 * @author LinHuiBa-YanAn
 * @date 2023/1/2 21:32
 */
@Slf4j
@Service
public class BaseMsgConsumerImpl implements BaseMsgConsumer {

	@Value("${config.rocket-mq.name-server-addr}")
	private String nameSrvAddr;

	@Override
	public void pushConsumerMsgByClustering(String consumerGroup, String topic, String tag,  MessageListener messageListener) {
		try {
			// 1. 创建消费者Consumer，指定消费者组名
			DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
			// 2. 指定NameServer地址
			consumer.setNamesrvAddr(nameSrvAddr);
			// 3. 订阅主题Topic和Tag
			consumer.subscribe(topic, tag);
			// 设定消费模式：负载均衡模式
			consumer.setMessageModel(MessageModel.CLUSTERING);
			// 4. 设置回调函数，处理消息
			consumer.registerMessageListener(messageListener);
			// 5. 启动消费者Consumer
			consumer.start();
		} catch (MQClientException e) {
			log.error("消费消息失败，e:{}", e);
			e.printStackTrace();
		}
	}

	@Override
	public void pushConsumerMsgByBroadcasting(String consumerGroup, String topic, String tag,  MessageListener messageListener) {
		try {
			// 1. 创建消费者Consumer，指定消费者组名
			DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
			// 2. 指定NameServer地址
			consumer.setNamesrvAddr(nameSrvAddr);
			// 3. 订阅主题Topic和Tag
			consumer.subscribe(topic, tag);
			// 设定消费模式：广播方式
			consumer.setMessageModel(MessageModel.BROADCASTING);
			// 4. 设置回调函数，处理消息
			consumer.registerMessageListener(messageListener);
			// 5. 启动消费者Consumer
			consumer.start();
		} catch (MQClientException e) {
			log.error("消费消息失败，e:{}", e);
			e.printStackTrace();
		}
	}
}
